package com.wh.weibo.service.impl;

import com.alibaba.fastjson.JSON;
import com.wh.weibo.facade.TopicService;
import com.wh.weibo.facade.WeiboService;
import com.wh.weibo.mapper.FeedMapper;
import com.wh.weibo.mapper.UserTopicFollowMapper;
import com.wh.weibo.mapper.UserWeiboLikeMapper;
import com.wh.weibo.mapper.WeiboMapper;
import com.wh.weibo.model.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author WeiHong
 * @date 2019/7/9 14:30
 */
@Service("weiboService")
public class WeiboServiceImpl implements WeiboService {

    @Resource
    private WeiboMapper weiboMapper;

    @Resource
    private UserWeiboLikeMapper userWeiboLikeMapper;

    @Resource
    private TopicService topicService;

    @Resource
    private FeedMapper feedMapper;

    @Resource
    private UserTopicFollowMapper userTopicFollowMapper;


    private DefaultMQProducer defaultMQProducer;

    @PostConstruct
    public void init() throws MQClientException {
        defaultMQProducer = new DefaultMQProducer("default_group");
        defaultMQProducer.setNamesrvAddr("192.168.247.131:9876");
        defaultMQProducer.start();

        DefaultMQPushConsumer consumerFeed = new DefaultMQPushConsumer("default_group");
        consumerFeed.setNamesrvAddr("192.168.247.131:9876");
        consumerFeed.subscribe("TopicTrade","TagA");
        consumerFeed.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                msgs.forEach(msg->{
                    String jsonString = new String(msg.getBody());
                    Weibo weibo = JSON.parseObject(jsonString,Weibo.class);
                    createFeed(weibo);
                });
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumerFeed.start();

        DefaultMQPushConsumer consumerDiscuss = new DefaultMQPushConsumer("default_group2");
        consumerDiscuss.setNamesrvAddr("192.168.247.131:9876");
        consumerDiscuss.subscribe("TopicTrade","TagB");
        consumerDiscuss.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                msgs.forEach(msg->{
                    String jsonString = new String(msg.getBody());
                    Weibo weibo = JSON.parseObject(jsonString,Weibo.class);
                    if(weibo.getTopicId() != null && weibo.getTopicId().intValue() != 0){
                        topicService.increaseDiscussCount(weibo.getTopicId());
                    }
                });
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumerDiscuss.start();
    }



    @Override
    public ServiceResult<Weibo> selectByPrimaryKey(Integer id) {
        ServiceResult result = new ServiceResult();
        if (null == id) {
            result.setSuccess(false);
            result.setErrorMsg("请求参数不能为空");
            return result;
        }
        Weibo weibo = weiboMapper.selectByPrimaryKey(id);
        if (null == weibo) {
            result.setSuccess(false);
            result.setErrorMsg("请求的微博不存在");
            return result;
        }
        result.setSuccess(true);
        result.setData(weibo);
        return result;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public ServiceResult<Weibo> create(Weibo weibo) {
        ServiceResult<Weibo> result = new ServiceResult<>();
        //判断参数是否合法
        if (StringUtils.isAnyBlank(weibo.getContent(), weibo.getImgUrl()) || null == weibo.getUserId()) {
            result.setSuccess(false);
            result.setErrorMsg("入参不合法");
            //操作切面回滚事务
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return result;
        }
        weiboMapper.insertSelective(weibo);
        //若微博关联了对应的话题，则需要对应的话题讨论数量+1
        if (null != weibo.getTopicId() && 0 != weibo.getTopicId().intValue()) {

            /*topicService.increaseDiscussCount(weibo.getTopicId());*/
            try {
                //使用异步化方式
                asyncIncreaseTopicCount(weibo);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (MQClientException e) {
                e.printStackTrace();
            } catch (RemotingException e) {
                e.printStackTrace();
            } catch (MQBrokerException e) {
                e.printStackTrace();
            }
        }


        result.setData(weibo);

        //事务提交完成后，创建对应的feed流的操作
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
            @Override
            public void afterCommit() {
                //createFeed(weibo);
                try {
                    //使用异步划方式创建对应的feed流的操作
                    sendFeedMsg(weibo);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (MQClientException e) {
                    e.printStackTrace();
                } catch (RemotingException e) {
                    e.printStackTrace();
                } catch (MQBrokerException e) {
                    e.printStackTrace();
                }
            }
        });
        return result;
    }

    /**
     * 使用异步划方式创建对应的feed流的操作
     * @param weibo
     * @throws InterruptedException
     * @throws MQClientException
     * @throws RemotingException
     * @throws MQBrokerException
     */
    protected void sendFeedMsg(Weibo weibo) throws InterruptedException, MQClientException, RemotingException, MQBrokerException {
        Message msg = new Message("TopicTrade","TagA", JSON.toJSON(weibo).toString().getBytes(Charset.forName("UTF-8")));
        defaultMQProducer.send(msg);

    }

    /**
     * 讨论数使用异步化
     *
     * @param weibo
     * @throws InterruptedException
     * @throws MQClientException
     * @throws RemotingException
     * @throws MQBrokerException
     */
    protected void asyncIncreaseTopicCount(Weibo weibo) throws InterruptedException, MQClientException, RemotingException, MQBrokerException {
        Message msg = new Message("TopicTrade","TagB", JSON.toJSON(weibo).toString().getBytes(Charset.forName("UTF-8")));
        defaultMQProducer.send(msg);

    }

    /**
     * 创建feed信息的操作
     * @param weibo
     */
    protected void createFeed(Weibo weibo){
        if (null != weibo.getTopicId() && 0 != weibo.getTopicId().intValue()){
            List<UserTopicFollow> follows = userTopicFollowMapper.listFollowByTopicId(weibo.getTopicId());
            follows.forEach(userTopicFollow -> {
                Feed feed = new Feed();
                feed.setCreateTime(weibo.getCreateTime());
                feed.setUserId(userTopicFollow.getUserId());
                feed.setLinkId(weibo.getId());
                feed.setType(1);
                try {
                    feedMapper.insert(feed);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public ServiceResult like(Integer userId, Integer weiboId) {
        ServiceResult result = new ServiceResult();
        if (null == userId || null == weiboId) {
            result.setSuccess(false);
            result.setErrorMsg("入参不合法");
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return result;
        }
        Weibo weibo = weiboMapper.selectByPrimaryKey(weiboId);
        if (null == weibo) {
            result.setSuccess(false);
            result.setErrorMsg("所点赞的微博不存在");
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return result;
        }

        try {
            UserWeiboLike like = new UserWeiboLike();
            like.setUserId(userId);
            like.setWeiboId(weiboId);
            userWeiboLikeMapper.insertSelective(like);
        } catch (DuplicateKeyException ex) {
            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
            return result;
        }
        weiboMapper.increaseLikeCount(weiboId);
        return result;
    }

    @Override
    public ServiceResult<List<Weibo>> selectList() {
        ServiceResult<List<Weibo>> result = new ServiceResult<>();
        List<Weibo> list = weiboMapper.selectList();
        result.setData(list);
        return result;
    }



    @Override
    public ServiceResult<List<Weibo>> selectListByTopicId(Integer topicId) {
        ServiceResult<List<Weibo>> result = new ServiceResult<>();
        if (null == topicId || 0 == topicId.intValue()){
            result.setSuccess(false);
            result.setErrorMsg("入参错误");
            return result;
        }
        List<Weibo> list = weiboMapper.selectListByTopicId(topicId);
        result.setData(list);
        return result;
    }

    @Override
    public ServiceResult<List<Weibo>> selectListByFeed(Integer userId) {
        ServiceResult<List<Weibo>> serviceResult= new ServiceResult<>();
        List<Feed> feeds = feedMapper.selectList(userId);
        if (feeds.size() == 0 ){
            serviceResult.setData(new ArrayList<>());
            return serviceResult;
        }

        List<Integer> weiboIds = feeds.stream().map(feed -> {
            return feed.getLinkId();
        }).collect(Collectors.toList());

        List<Weibo> weiboList = weiboMapper.selectListByWeiBoIds(weiboIds);
        serviceResult.setData(weiboList);
        return serviceResult;
    }
}
